import asyncio
import base64
import functools
import hashlib
import logging
import pickle
from typing import Any, Awaitable, Callable, Union

from redis.asyncio import Redis

# 配置日志
logger = logging.getLogger("AsyncRedisCache")


class AsyncRedisCache:
    """
    基于 Redis 的异步缓存装饰器类，支持缓存、更新、清除等操作。

    特性：
        - 支持标准的缓存模式（@cache）、强制更新模式（@cache_put）和清除模式（@cache_evict）
        - 自动处理序列化/反序列化，支持复杂对象
        - 灵活的缓存键生成策略，避免键冲突
        - 完善的错误处理，缓存操作失败不会影响主业务流程
        - 支持批量清除和指定参数清除

    注意：
        - 如果 Redis 客户端设置 `decode_responses=True`，会自动使用 base64 编码处理二进制数据
        - 所有缓存操作失败仅记录警告，不会中断原函数执行
    """

    def __init__(self, redis_client: Redis, key_prefix: str = "cache"):
        """
        初始化缓存管理器。

        Args:
            redis_client: 异步 Redis 客户端实例
            key_prefix: 缓存键的统一前缀，用于命名空间隔离

        Raises:
            TypeError: 如果 redis_client 不是 Redis 实例
        """
        if not isinstance(redis_client, Redis):
            raise TypeError("redis_client must be an instance of redis.asyncio.Redis")

        self.redis = redis_client
        self.key_prefix = key_prefix.strip(":")
        self.decode_responses = getattr(redis_client, "decode_responses", False)

        logger.info(
            f"AsyncRedisCache initialized with prefix: {self.key_prefix}, " f"decode_responses: {self.decode_responses}"
        )

    def _serialize_value(self, obj: Any) -> Union[bytes, str]:
        """
        序列化对象为 Redis 可存储的格式。

        Args:
            obj: 需要序列化的对象

        Returns:
            序列化后的数据（bytes 或 str）

        Raises:
            pickle.PicklingError: 序列化失败时抛出
        """
        try:
            serialized = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
            if self.decode_responses:
                # 如果 Redis 配置为解码响应，使用 base64 编码避免字符串问题
                return base64.b64encode(serialized).decode("utf-8")
            return serialized
        except (pickle.PicklingError, TypeError) as e:
            logger.error(f"Failed to serialize object: {e}")
            raise pickle.PicklingError(f"Object serialization failed: {e}")

    def _deserialize_value(self, data: Union[bytes, str, None]) -> Any:
        """
        反序列化 Redis 中的数据。

        Args:
            data: 从 Redis 获取的原始数据

        Returns:
            反序列化后的对象

        Raises:
            pickle.UnpicklingError: 反序列化失败时抛出
        """
        if data is None:
            return None

        try:
            if self.decode_responses and isinstance(data, str):
                # 解码 base64 字符串
                data = base64.b64decode(data.encode("utf-8"))
            return pickle.loads(data)
        except (pickle.UnpicklingError, TypeError, ValueError) as e:
            logger.error(f"Failed to deserialize data: {e}")
            raise pickle.UnpicklingError(f"Data deserialization failed: {e}")

    def _generate_key(self, func: Callable, *args, **kwargs) -> str:
        """
        生成唯一缓存键。

        策略：
          1. 使用函数名、参数生成唯一标识
          2. 对不可序列化参数有回退方案
          3. 使用 MD5 哈希避免过长的键名

        Args:
            func: 被装饰的函数
            args: 位置参数
            kwargs: 关键字参数

        Returns:
            唯一缓存键字符串
        """
        # 创建参数的稳定表示（忽略顺序）
        key_parts = [
            func.__module__ or "",  # 模块名
            func.__qualname__,  # 函数全名
        ]

        _key_prefix = f"{self.key_prefix}:{':'.join(key_parts)}"
        try:
            # 处理位置参数
            for arg in args:
                key_parts.append(repr(arg))

            # 处理关键字参数（按键名排序保证稳定）
            for key in sorted(kwargs.keys()):
                key_parts.append(f"{key}={repr(kwargs[key])}")

            key_string = ":".join(key_parts)
            key_hash = hashlib.md5(key_string.encode("utf-8")).hexdigest()

            return f"{_key_prefix}:{key_hash}"

        except Exception as e:
            logger.warning(f"Key generation failed for {func.__qualname__}: {e}. Using fallback method.")
            # 回退方案：简化处理
            fallback_string = f"{_key_prefix}:{len(args)}:{len(kwargs)}"
            fallback_hash = hashlib.md5(fallback_string.encode("utf-8")).hexdigest()
            return f"{_key_prefix}:fallback:{fallback_hash}"

    async def _safe_redis_operation(self, operation: Callable, *args, **kwargs) -> Any:
        """
        安全的 Redis 操作包装器，捕获所有异常避免影响主流程。

        Args:
            operation: Redis 操作方法
            *args: 操作参数
            **kwargs: 操作关键字参数

        Returns:
            操作结果，失败时返回 None
        """
        try:
            return await operation(*args, **kwargs)
        except Exception as e:
            logger.warning(f"Redis operation failed: {e}")
            return None

    def cache(self, ttl: int = 300):
        """
        缓存装饰器：先查缓存，命中则返回；未命中则执行函数并缓存结果。

        使用示例：
            @cache_manager.cache(ttl=600)
            async def get_data(key: str) -> dict:
                return await fetch_from_db(key)

        Args:
            ttl: 缓存过期时间（秒），默认 5 分钟

        Returns:
            装饰后的异步函数
        """

        def decorator(func: Callable[..., Awaitable[Any]]):
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                cache_key = self._generate_key(func, *args, **kwargs)

                # 1. 尝试从缓存读取
                cached_result = await self._safe_redis_operation(self.redis.get, cache_key)

                if cached_result is not None:
                    try:
                        result = self._deserialize_value(cached_result)
                        logger.debug(f"Cache hit for {func.__qualname__} with key: {cache_key}")
                        return result
                    except Exception as e:
                        logger.warning(f"Cache deserialization failed for {cache_key}: {e}")
                        # 反序列化失败，继续执行函数

                # 2. 缓存未命中或无效，执行原函数
                logger.debug(f"Cache miss for {func.__qualname__} with key: {cache_key}")
                result = await func(*args, **kwargs)

                # 3. 异步写入缓存（不阻塞主流程）
                asyncio.create_task(self._set_cache(cache_key, result, ttl))

                return result

            # 添加缓存管理方法
            wrapper.delete_cache = self._create_delete_cache_method(func)
            wrapper.clear_all_cache = self._create_clear_all_cache_method(func)
            wrapper.get_cache_key = self._create_get_cache_key_method(func)

            return wrapper

        return decorator

    def cache_put(self, ttl: int = 300):
        """
        强制更新缓存：总是执行函数并更新缓存结果。

        适用于数据更新场景，保证缓存与数据源同步。

        Args:
            ttl: 缓存过期时间（秒）

        Returns:
            装饰后的异步函数
        """

        def decorator(func: Callable[..., Awaitable[Any]]):
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                result = await func(*args, **kwargs)
                cache_key = self._generate_key(func, *args, **kwargs)
                # 异步更新缓存
                asyncio.create_task(self._set_cache(cache_key, result, ttl))
                return result

            return wrapper

        return decorator

    def cache_evict(self, before_invocation: bool = True):
        """
        清除缓存：在执行函数前后删除相关缓存。

        Args:
            before_invocation: 是否在函数执行前清除缓存（True：执行前，False：执行后）

        Returns:
            装饰后的异步函数
        """

        def decorator(func: Callable[..., Awaitable[Any]]):
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                cache_key = self._generate_key(func, *args, **kwargs)
                if before_invocation:
                    # 执行前清除缓存
                    await self._safe_redis_operation(self.redis.delete, cache_key)
                result = await func(*args, **kwargs)
                if not before_invocation:
                    # 执行后清除缓存
                    await self._safe_redis_operation(self.redis.delete, cache_key)
                return result

            return wrapper

        return decorator

    async def _set_cache(self, key: str, value: Any, ttl: int) -> None:
        """异步设置缓存，错误处理已内置"""
        try:
            serialized_value = self._serialize_value(value)
            await self._safe_redis_operation(self.redis.setex, key, ttl, serialized_value)
            logger.debug(f"Cache set for key: {key} (TTL: {ttl}s)")
        except Exception as e:
            logger.warning(f"Failed to set cache for key {key}: {e}")

    def _create_delete_cache_method(self, func: Callable) -> Callable:
        """创建删除指定参数缓存的方法"""

        async def delete_cache(*args, **kwargs):
            cache_key = self._generate_key(func, *args, **kwargs)
            deleted = await self._safe_redis_operation(self.redis.delete, cache_key)
            if deleted:
                logger.info(f"Cache deleted for key: {cache_key}")
            else:
                logger.info(f"Cache key not found: {cache_key}")
            return deleted

        return delete_cache

    def _create_clear_all_cache_method(self, func: Callable) -> Callable:
        """创建清除函数所有缓存的方法"""

        async def clear_all_cache():
            pattern = f"{self.key_prefix}:{func.__qualname__}:*"
            keys = await self._safe_redis_operation(self.redis.keys, pattern)
            if keys:
                deleted_count = await self._safe_redis_operation(self.redis.delete, *keys)
                logger.info(f"Cleared {deleted_count} cache entries for {func.__qualname__}")
                return deleted_count
            else:
                logger.info(f"No cache entries found for {func.__qualname__}")
                return 0

        return clear_all_cache

    def _create_get_cache_key_method(self, func: Callable) -> Callable:
        """创建获取缓存键的方法（用于调试）"""

        def get_cache_key(*args, **kwargs):
            return self._generate_key(func, *args, **kwargs)

        return get_cache_key


# ==================== 使用示例 ====================


# async def main():
#     """使用示例和测试"""
#     # 创建 Redis 客户端（支持两种模式）
#     redis_client_bytes = Redis(host="localhost", port=6379, db=0, decode_responses=False)
#     redis_client_str = Redis(host="localhost", port=6379, db=1, decode_responses=True)

#     # 创建缓存管理器
#     cache_manager_bytes = AsyncRedisCache(redis_client_bytes, "myapp_bytes")
#     cache_manager_str = AsyncRedisCache(redis_client_str, "myapp_str")

#     # 使用字节模式缓存管理器
#     @cache_manager_bytes.cache(ttl=60)
#     async def get_user_data(user_id: int, category: str = "profile") -> dict:
#         """模拟数据库查询"""
#         logger.info(f"Executing get_user_data for user {user_id}, category {category}")
#         await asyncio.sleep(0.1)
#         return {
#             "user_id": user_id,
#             "category": category,
#             "data": f"user_data_{user_id}_{category}",
#             "nested": {"level": 1, "items": [1, 2, 3]},
#         }

#     @cache_manager_bytes.cache_put(ttl=120)
#     async def update_user_data(user_id: int, data: dict) -> dict:
#         """更新用户数据（总是更新缓存）"""
#         logger.info(f"Updating user {user_id} data")
#         await asyncio.sleep(0.1)
#         return {"user_id": user_id, **data, "updated": True}

#     @cache_manager_bytes.cache_evict()
#     async def delete_user_data(user_id: int):
#         """删除用户数据（清除缓存）"""
#         logger.info(f"Deleting user {user_id} data")
#         await asyncio.sleep(0.1)
#         return {"status": "deleted", "user_id": user_id}

#     # 测试缓存功能
#     print("=== Testing @cache ===")
#     result1 = await get_user_data(1, "profile")
#     print("Result 1:", result1)

#     result2 = await get_user_data(1, "profile")  # 应命中缓存
#     print("Result 2 (cached):", result2)

#     # 测试缓存管理方法
#     cache_key = get_user_data.get_cache_key(1, "profile")
#     print(f"Cache key for user 1: {cache_key}")

#     # 测试清除缓存
#     await get_user_data.delete_cache(1, "profile")

#     result3 = await get_user_data(1, "profile")  # 应重新执行
#     print("Result 3 (after delete):", result3)

#     # 清理资源
#     await redis_client_bytes.close()
#     await redis_client_str.close()


# if __name__ == "__main__":
#     asyncio.run(main())
